Kevin-Blog

频率组件


只要继承了 APIView 就可以使用频率组件

组件的执行顺序:  认证 -> 权限 -> 频率

1. 频率类的固定写法

# rf_throttle.py

class 频率类的类名(object):

    def allow_request(self, request, view):

 # 频率的逻辑代码

        if True / False:
            return True
        else:
            return False

    def wait(self):
# 钩子函数（即: 回调函数）
        # 当超过设置的访问频率时调用（即: 当allow_request返回False时调用）
        # 返回需要等待的时间

        return '时间字符串'

2. 局部频率 -> 自己编写频率类

  • 局部频率的配置（即: throttle_classes = [频率类, 频率类]）会覆盖掉该视图类在全局频率中的配置

  • throttle_classes = [频率类, 频率类]

# rf_throttle.py

import time

VISIT_RECORD = {
# '127.0.0.1:8080':[最新的时间，……，最初的时间]
}


class MyThrottle(object):
 """
    一分钟允许访问5次
    原理:
        拿到用户IP地址
        编写访问记录 {'ip地址':[最新的时间，……，最初的时间]}
        确保用户所对应的记录列表的最新时间和最初时间的时间差 <= 规定时间
        判断用户所对应的记录列表的长度 < 允许访问次数
    """

    allow_time = 60  # 时间
    visits = 5  # 允许访问次数

    def __init__(self):
        self.history_list = []

    def allow_request(self, request, view):
"""
        :param request: APIView 的 request
        :param view: 视图类的实例化对象 -> APIView 类中的 self
        :return: True/False
        """

# 获取用户的IP地址
        ip = request.META.get('REMOTE_ADDR', '')
        if ip not in VISIT_RECORD:
            VISIT_RECORD[ip] = [time.time()]
        else:
            history = VISIT_RECORD[ip]

            if len(history) <= self.visits:
                history.insert(0, time.time())  # 将最新的时间添加到列表的第一位

            self.history_list = history  # 因为 history 是列表，所以 self.history_list = history 是引用关系，没有进行深度拷贝，所以下面 history.pop() 执行 self.history_list 无需重新赋值

            while time.time() - history[-1] > self.allow_time:
 # 确保列表时间是允许的范围内（即: 列表的时间差不能超过60秒）
                # 循环计算最新的时间和最初的时间是否在60秒，如果不是删除则删除最初的时间
                history.pop()

# 判断列表的长度有没有超过允许访问次数
            if len(history) > self.visits:
                return False

        return True

    def wait(self):
# 钩子函数（即: 回调函数）
        # 当超过设置的访问频率时调用（即: 当allow_request返回False时调用）
        # 返回值等待时间
        return self.allow_time - (time.time() - self.history_list[-1])

# views.py

from .models import *
from .serializer import *
from .rf_auth import *
from .rf_permission import *
from .rf_throttle import *

from rest_framework import viewsets
from rest_framework import exceptions


# 修改频率组件的错误信息
class MyExceptions(exceptions.Throttled):
    default_detail = '连接次数过多'
    extra_detail_plural = extra_detail_singular = '请在{wait}秒后访问'

    def __init__(self, wait=None, detail=None, code=None):
        super().__init__(wait=wait, detail=detail, code=code)


class BookViewSet(viewsets.ModelViewSet):
    authentication_classes = [TokenAuth]  # 局部认证，只作用于当前视图
    permission_classes = [SVIPPermission]  # 局部权限，只作用于当前视图
throttle_classes = [MyThrottle]  # 局部频率，只作用于当前视图

    queryset = Book.objects.all()
    serializer_class = BookSerializers

# 修改频率组件的错误信息
    def throttled(self, request, wait):
        raise MyExceptions(wait=wait)

2. 全局频率 -> 自己编写频率类

  • 全局频率作用于所有视图类

# rf_throttle.py

import time

VISIT_RECORD = {
# '127.0.0.1:8080':[最新的时间，……，最初的时间]
}


class MyThrottle(object):
 """
    一分钟允许访问5次
    原理:
        拿到用户IP地址
        编写访问记录 {'ip地址':[最新的时间，……，最初的时间]}
        确保用户所对应的记录列表的最新时间和最初时间的时间差 <= 规定时间
        判断用户所对应的记录列表的长度 < 允许访问次数
    """

    allow_time = 60  # 时间
    visits = 5  # 允许访问次数

    def __init__(self):
        self.history_list = []

    def allow_request(self, request, view):
"""
        :param request: APIView 的 request
        :param view: 视图类的实例化对象 -> APIView 类中的 self
        :return: True/False
        """

# 获取用户的IP地址
        ip = request.META.get('REMOTE_ADDR', '')
        if ip not in VISIT_RECORD:
            VISIT_RECORD[ip] = [time.time()]
        else:
            history = VISIT_RECORD[ip]

            if len(history) <= self.visits:
                history.insert(0, time.time())  # 将最新的时间添加到列表的第一位

            self.history_list = history  # 因为 history 是列表，所以 self.history_list = history 是引用关系，没有进行深度拷贝，所以下面 history.pop() 执行 self.history_list 无需重新赋值

            while time.time() - history[-1] > self.allow_time:
 # 确保列表时间是允许的范围内（即: 列表的时间差不能超过60秒）
                # 循环计算最新的时间和最初的时间是否在60秒，如果不是删除则删除最初的时间
                history.pop()

# 判断列表的长度有没有超过允许访问次数
            if len(history) > self.visits:
                return False

        return True

    def wait(self):
# 钩子函数（即: 回调函数）
        # 当超过设置的访问频率时调用（即: 当allow_request返回False时调用）
        # 返回值等待时间
        return self.allow_time - (time.time() - self.history_list[-1])

# settings.py

REST_FRAMEWORK = {
 # "DEFAULT_AUTHENTICATION_CLASSES": ["app01.rf_auth.TokenAuth"],  # 全局认证
    # "DEFAULT_PERMISSION_CLASSES": ["app01.rf_permission.SVIPPermission"]  # 全局权限
    "DEFAULT_THROTTLE_CLASSES": ["app01.rf_throttle.MyThrottle"]  # 全局频率
}

  • 不进行频率认证的视图类设置

    • 如果设置了全局频率，但是又想某些视图不进行频率认证，那么可以在不想进行频率认证的视图类中设置 throttle_classes = []，因为在 rest-framework 源码中 局部频率的配置 会覆盖掉 全局频率的配置

# views.py

import hashlib
import time

from django.shortcuts import render, HttpResponse
from rest_framework.views import APIView
from rest_framework.response import Response
from .models import *
from .serializer import *
from .rf_auth import *

from rest_framework import viewsets


class BookViewSet(viewsets.ModelViewSet):
    queryset = Book.objects.all()
    serializer_class = BookSerializers


class PublishViewSet(viewsets.ModelViewSet):
    queryset = Publish.objects.all()
    serializer_class = PublishSerializers


# 获取以用户名作为盐，以当前时间作为加密内容的字符串token
def get_token_str(username):
    now_time = str(time.time())
    md5 = hashlib.md5(bytes(username, encoding="utf8"))  # 加盐，使用用户名作为盐，保证token的唯一性
    md5.update(bytes(now_time, encoding='utf-8'))
    return md5.hexdigest()


# 登陆，并且返回当前用户的token
class LoginView(APIView):
throttle_classes = [] # 在全局频率下，当前视图不进行频率认证

    def post(self, request):
        username = request.data.get('username')
        password = request.data.get('password')
        user = User.objects.filter(username=username, password=password).first()
        ret = {
            'status': 0,
            'msg': None
        }
        if user:
            token_str = get_token_str(user.username)  # 获取Token值
# 每次登陆成功后都要修改当前用户的token值
            Token.objects.update_or_create(user=user, defaults={'token': token_str})  # 如果有 user=user 这条数据，那么就修改 token 值，否则就进行添加
            ret['status'] = 1
            ret['token'] = token_str
        else:
            ret['msg'] = '登陆失败'

        return Response(ret)

4. 局部频率 ->  使用 rest-framework 所提供的频率类

# rf_throttle.py

from rest_framework.throttling import SimpleRateThrottle


class MyThrottle(SimpleRateThrottle):
    rate = '5/m'  # 每分钟允许访问5次

# 获取用户IP地址
    def get_cache_key(self, request, view):
        return self.get_ident(request)

# views.py

from .rf_throttle import *
from rest_framework import viewsets


class BookViewSet(viewsets.ModelViewSet):
throttle_classes = [MyThrottle]
    queryset = Book.objects.all()
    serializer_class = BookSerializers

5. 全局频率 -> 使用 rest-framework 所提供的频率类

# rf_throttle.py

from rest_framework.throttling import SimpleRateThrottle


class MyThrottle(SimpleRateThrottle):
    scope = 'visit_rate'  # 设置全局频率的属性名

# 获取用户IP地址
    def get_cache_key(self, request, view):
        return self.get_ident(request)

# settings.py

REST_FRAMEWORK = {
# "DEFAULT_AUTHENTICATION_CLASSES": ["app01.rf_auth.TokenAuth"],  # 全局认证
    # "DEFAULT_PERMISSION_CLASSES": ["app01.rf_permission.SVIPPermission"]  # 全局权限
    "DEFAULT_THROTTLE_CLASSES": ["app01.rf_throttle.MyThrottle"],  # 全局频率
    "DEFAULT_THROTTLE_RATES": {
        'visit_rate': '5/m'  # 每分钟允许访问5次
    }
}



←